package com.baishancloud.log.douyin.demand2.subtask

import com.baishancloud.log.common.util.{IpInfoUtil, StringUtil, TimeUtil}
import com.baishancloud.log.douyin.{kafkaSourceJh, mallardName2}
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

import java.util
import java.util.Objects

/**
 * 由于头条  域名数据从金华地区切换到石家庄地区 kafka 集群，因此抖音的请求需要两个地方进行写入，故在原来的基础上，在 tag 标签中增加 kafka_source 字段来群是消费的哪个地区的 kafka 集群数据。<br>
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 16:01
 */
class LogParse2Sub(kafkaSource:String) extends ProcessFunction[Origin, Result2Sub] {

  override def processElement(value: Origin, ctx: ProcessFunction[Origin, Result2Sub]#Context, out: Collector[Result2Sub]): Unit = {
    // a 表示需求2中的 a 维度
    var a: String = "-1"
    try {
      val aValue: util.List[String] = StringUtil.split(StringUtil.split(StringUtil.split(value.url(), '?').get(1), '&').get(0), '=')
      if (aValue.size() == 2 && Objects.equals(aValue.get(0), "a")) {
        a = aValue.get(1)
      }
    } catch {
      case _: Exception =>
    }
    out.collect(
      Result2Sub(
        TimeUtil.timeAtLong(value.datetimeOfUTC().toEpochSecond, 60),
        60,
        mallardName2,
        "",
        1L,
        Tags2Sub(
          value.domain(),
          IpInfoUtil.core(value.ip()).isp,
          a,
          kafkaSource
        )
      )
    )
  }
}
